package com.atguigu.gmall.realtime.app.function;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.ververica.cdc.debezium.DebeziumDeserializationSchema;
import io.debezium.data.Envelope;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.util.Collector;
import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;

import java.util.List;

public class CustomerDeserialization implements DebeziumDeserializationSchema<String> {

    /**
     * 封装的数据格式
     * ｛
     * “database”：“”，
     * “tableName”：“”，
     * “type”：“c u d”，
     * “before”：“{"":"","":""......}”
     * "after":{"":"","":""......}
     * //"ts",1561243212
     * ｝
     *
     * @param sourceRecord
     * @param collector
     * @throws Exception
     */
    @Override
    public void deserialize(SourceRecord sourceRecord, Collector<String> collector) throws Exception {

        //1.创建JSON对象用于存储最终数据
        JSONObject result = new JSONObject();

//        2.获取库名&表名
        String topic = sourceRecord.topic();
        String[] fields = topic.split("\\.");
        String database = fields[1];
        String tableName = fields[2];

        Struct value = (Struct) sourceRecord.value();
//        3、获取"before"数据
        Struct before = value.getStruct("before");
//        创建一个before json
        JSONObject beforeJson = new JSONObject();

//        判断befor不为null 防止 空指针异常
        if (before != null) {
            //        循环取出写入数据 可以通过 Struct的 Schema 信息
            Schema beforeSchema = before.schema();
            List<Field> beforeFields = beforeSchema.fields();
                for (int i = 0; i < beforeFields.size(); i++) {
                    //            定位到具体的字段
                    Field field = beforeFields.get(i);
    //            定位到具体的字段获取 字段值
                    Object beforeValue = before.get(field);
    //            把结果放入before json里面
                    beforeJson.put(field.name(), beforeValue);
                }
        }


//        4.获取"after"数据

        Struct after = value.getStruct("after");
//        创建一个before json
        JSONObject afterJson = new JSONObject();

//        判断befor不为null 防止 空指针异常
        if (after != null) {
            //        循环取出写入数据 可以通过 Struct的 Schema 信息
            Schema afterSchema = after.schema();
            List<Field> afterFields = afterSchema.fields();
            for (int i = 0; i < afterFields.size(); i++) {
                //            定位到具体的字段
                Field field = afterFields.get(i);
                //            定位到具体的字段获取 字段值
                Object afterValue = after.get(field);
                //            把结果放入before json里面
                afterJson.put(field.name(), afterValue);
            }
        }

        //5.获取操作类型
        Envelope.Operation operation = Envelope.operationFor(sourceRecord);
        String type = operation.toString().toLowerCase();
//        根据后面动态分流映射表 为了跟canal maxwell 统一
        if("create".equals(type)){
            type = "insert";
        }

        //6.将字段写入JSON对象
        result.put("database",database);
        result.put("tableName",tableName);
        result.put("before",beforeJson);
        result.put("after",afterJson);
        result.put("type",type);

//        7.输出数据
        collector.collect(result.toJSONString());
    }

    @Override
    public TypeInformation<String> getProducedType() {
        return BasicTypeInfo.STRING_TYPE_INFO;
    }
}
